package com.hellorxjava.demo;


import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Maybe;
import io.reactivex.rxjava3.core.MaybeObserver;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;

/**
 * throttleWithTimeout 通过时间来限流 如果在设定好的时间结束前又接收到源Observable发送的新数据 那么这个限流会走向极端只接收最后一个数据
 * distinct 过滤重复的数据
 * distinctUntilChanged 过滤连续重复的数据 可能仍然有重复
 * elementAt 取出第几个发送的数据
 * filter true的话就会被发送出来 否则就会被过滤掉
 * first 第一条数据或者满足条件的第一条数据
 * last 第一条数据或者满足条件的最后一条数据
 * 将Observable转化为BlockingObservable 然后可以用first操作符来取满足条件的第一个数据
 *
 * skip 可以过滤掉前n项
 * take操作符则只取前n项
 *
 * sample取规定时间内的最后一个数据
 * throttleFirst取的是规定时间内的第一个数据
 */
public class RXJavaDemo6 {
    public static void main(String[] args) {
        throttleWithTimeOutObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("throttleWithTimeOutObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }
        });

        debounceObserver().subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("debounceObserver onSubscribe");
            }

            @Override
            public void onNext(@NonNull Long aLong) {
                System.out.println("debounceObserver onNext " + aLong);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("debounceObserver onComplete ");
            }
        });

        distinctObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("distinctObserver onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("debounceObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("debounceObserver onComplete ");
            }
        });

        distinctUtilChangeObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("distinctUtilChangeObserver onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("distinctUtilChangeObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("distinctUtilChangeObserver onComplete");
            }
        });

        elementAtObserver().subscribe(new MaybeObserver<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("elementAtObserver onSubscribe");
            }

            @Override
            public void onSuccess(@NonNull Integer integer) {
                System.out.println("elementAtObserver onSuccess " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("elementAtObserver onComplete ");
            }
        });

        filterObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("filterObserver onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("filterObserver onSuccess " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("filterObserver onComplete ");
            }
        });

         Integer i = Observable.just(0, 1, 2, 3, 4, 5).blockingFirst();
         System.out.println("Observable " + i);
        Observable.just(0,1,2,3,4,5).firstElement().subscribe(new MaybeObserver<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("onSubscribe ");
            }

            @Override
            public void onSuccess(@NonNull Integer integer) {
                System.out.println("firstElement onSuccess " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("onComplete ");
            }
        });

        skipObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("skipObserver onSubscribe ");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("skipObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("skipObserver onComplete ");
            }
        });

        takeObserver().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("takeObserver onSubscribe ");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                System.out.println("takeObserver onNext " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("takeObserver onComplete ");
            }
        });

        sampleObserver().subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("sampleObserver onSubscribe ");
            }

            @Override
            public void onNext(@NonNull Long aLong) {
                System.out.println("sampleObserver onNext " + aLong);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("sampleObserver onComplete ");
            }
        });

        throttleFirstObserver().subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("throttleFirstObserver onSubscribe ");
            }

            @Override
            public void onNext(@NonNull Long aLong) {
                System.out.println("throttleFirstObserver onNext " + aLong);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("throttleFirstObserver onComplete ");
            }
        });
    }

    private static Observable<Integer> createObserver() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
                for (int i = 0; i < 10; i++) {
                    if(!emitter.isDisposed()) {
                        emitter.onNext(i);
                    }
                    int sleep = 100;
                    if (i % 3 == 0) {
                        sleep =300;
                    }
                    try {
                        Thread.sleep(sleep);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                emitter.onComplete();
            }
        });
    }

    private static Observable<Integer> throttleWithTimeOutObserver(){
        return createObserver().throttleWithTimeout(200, TimeUnit.MILLISECONDS);
    }

    private static Observable<Long> debounceObserver(){
        return Observable.interval(1000, TimeUnit.MILLISECONDS)  //隔1秒发送数字
                .debounce(new Function<Long, ObservableSource<Long>>() {
                    @Override
                    public ObservableSource<Long> apply(Long aLong) throws Throwable {
                        return Observable.timer(aLong  % 2 * 1500, TimeUnit.MILLISECONDS); //延迟
                    }
                });
    }

    private static Observable<Integer> distinctObserver(){
        return Observable.just(1,2,3,4,5,5,4,3,2,1).distinct();
    }

    //过滤连续的重复数据 可能还是有重复的
    private static Observable<Integer> distinctUtilChangeObserver(){
        return Observable.just(1,2,3,4,5,5,4,3,2,1).distinctUntilChanged();
    }

    private static @NonNull Maybe<Integer> elementAtObserver(){
        return Observable.just(0,1,2,3,4,5).elementAt(2);
    }


    private static Observable<Integer> filterObserver(){
        return Observable.just(0,1,2,3,4,5).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Throwable {
                return integer < 3;
            }
        });
    }

    private static Observable<Integer> skipObserver(){
        return Observable.just(0,1,2,3,4,5).skip(2);
    }

    private static Observable<Integer> takeObserver(){
        return Observable.just(0,1,2,3,4,5).take(2);
    }

    private static Observable<Long> sampleObserver(){
        return Observable.interval(200, TimeUnit.MILLISECONDS).sample(1000,TimeUnit.MILLISECONDS);
    }

    private static Observable<Long> throttleFirstObserver(){
        return Observable.interval(200, TimeUnit.MILLISECONDS).throttleFirst(1000,TimeUnit.MILLISECONDS);
    }
}
